package cn._51doit.live.udf;

import cn._51doit.live.pojo.DataBean;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 将JSON字符串转成Bean的Function
 * 1.数据的转换
 * 2.问题数据的过滤
 */
public class JsonToBeanFunctionV2 extends ProcessFunction<Tuple2<String, String>, DataBean> {


    @Override
    public void processElement(Tuple2<String, String> value, Context ctx, Collector<DataBean> out) throws Exception {

        try {
            String id = value.f0;
            String line = value.f1;
            DataBean dataBean = JSON.parseObject(line, DataBean.class);
            dataBean.setId(id);
            out.collect(dataBean);
        } catch (Exception e) {
            //e.printStackTrace();
        }


    }
}
